package biz.console;

import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.ByteBufUtil;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.CodecException;
import io.netty.handler.codec.redis.*;
import io.netty.util.CharsetUtil;
import io.netty.util.ReferenceCountUtil;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import reactor.core.publisher.Flux;

import java.time.Duration;
import java.time.LocalDateTime;
import java.util.*;
import java.util.concurrent.*;

/**
 * @创建人 xiaojiancheng
 * @时间 2020/8/5
 * @描述 $k$
 */
@Slf4j
public class RedisClient {
    @Setter
    @Getter
    private String host;
    @Setter
    @Getter
    private int port;
    @Setter
    @Getter
    private String password;

    private Channel channel;
    Semaphore sem = new Semaphore(0);
    Object result = null;

    private static Map<Channel, RedisClient> clientMap = new ConcurrentHashMap<>();
    private static Map<String, RedisClient> keyMap = new ConcurrentHashMap<>();

    private LocalDateTime beatTime = LocalDateTime.now();

    public RedisClient(String host, int port, String password) {
        this.host = host;
        this.port = port;
        this.password = password;
    }

    static RedisClientHandler1 redisClientHandler;

    static Bootstrap b;
    static EventLoopGroup group;

    static {
        group = new NioEventLoopGroup(1,new ThreadPoolExecutor(1,1,10, TimeUnit.SECONDS,new LinkedBlockingDeque<>()));
        b = new Bootstrap();
        b.group(group)
                .channel(NioSocketChannel.class)
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel ch) throws Exception {
                        ChannelPipeline p = ch.pipeline();
                        p.addLast(new RedisDecoder());
                        p.addLast(new RedisBulkStringAggregator());
                        p.addLast(new RedisArrayAggregator());
                        p.addLast(new RedisEncoder());
                        redisClientHandler = new RedisClientHandler1();
                        p.addLast(redisClientHandler);
                    }
                });

        //  15分钟定时检测心跳 没有动作的链接会被释放掉
        Flux.interval(Duration.ofMillis(15))
                .subscribe(time -> {
                    List<String> keyList = new LinkedList<>();
                    LocalDateTime now = LocalDateTime.now();
                    keyMap.forEach((k,v) -> {
                        //  如果减去15分钟还比心跳时间大 那么退出
                        if(now.plusMinutes(-15).isAfter(v.beatTime)){
                            keyList.add(k);
                        }
                    });
                    //  关闭链接
                    for (String s : keyList) {
                        RedisClient remove = keyMap.remove(s);
                        if(remove != null){
                            //  关闭链接
                            remove.channel.close();
                            clientMap.remove(remove.channel);
                        }
                    }
                });
    }

    public String connect() throws Exception {
        channel = b.connect(host, port).sync().channel();
        clientMap.put(channel,this);
        String uuid = UUID.randomUUID().toString();
        keyMap.put(uuid,this);
        //  如果有密码
        if(password != null && !Objects.equals(password,"")){
            channel.writeAndFlush("auth "+password);
            sem.acquire();
            return uuid;
        }
        return uuid;
    }

    public Object write(String line) throws Exception{
        channel.writeAndFlush(line);
        sem.acquire();
        return result;
    }

    public void beat(){
        beatTime = LocalDateTime.now();
    }

    public static RedisClient get(String uuid){
        return keyMap.get(uuid);
    }


    private static class RedisClientHandler1 extends ChannelDuplexHandler {

        @Override
        public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) {
            String[] commands = ((String) msg).split("\\s+");
            List<RedisMessage> children = new ArrayList<RedisMessage>(commands.length);
            for (String cmdString : commands) {
                children.add(new FullBulkStringRedisMessage(ByteBufUtil.writeUtf8(ctx.alloc(), cmdString)));
            }
            RedisMessage request = new ArrayRedisMessage(children);
            ctx.write(request, promise);
        }

        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) {
            RedisMessage redisMessage = (RedisMessage) msg;
            Object result = response(redisMessage);
            RedisClient client = clientMap.get(ctx.channel());
            if(client != null){
                client.result = result;
                client.sem.release();
            }
            ReferenceCountUtil.release(redisMessage);
        }

        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
            cause.printStackTrace(System.err);
            clientMap.remove(ctx.channel());
        }


        private Object response(RedisMessage msg) {
            Object content_ = null;
            if (msg instanceof SimpleStringRedisMessage) {
                content_ = ((SimpleStringRedisMessage) msg).content();
            } else if (msg instanceof ErrorRedisMessage) {
                System.out.println(((ErrorRedisMessage) msg).content());
                content_ = ((ErrorRedisMessage) msg).content();
            } else if (msg instanceof IntegerRedisMessage) {
                System.out.println(((IntegerRedisMessage) msg).value());
                content_ = ((IntegerRedisMessage) msg).value();
            } else if (msg instanceof FullBulkStringRedisMessage) {
                System.out.println(getString((FullBulkStringRedisMessage) msg));
                content_ = getString(((FullBulkStringRedisMessage) msg));
            } else if (msg instanceof ArrayRedisMessage) {
                List<Object> list = new LinkedList<>();
                for (RedisMessage child : ((ArrayRedisMessage) msg).children()) {
                    list.add(response(child));
                }
                content_ = list;
            } else {
                throw new CodecException("unknown message type: " + msg);
            }
            return content_;
        }

        private String getString(FullBulkStringRedisMessage msg) {
            if (msg.isNull()) {
                return "(null)";
            }
            return msg.content().toString(CharsetUtil.UTF_8);
        }
    }
}
